Best Practice

Helper Utilities

Testplan provides helper functions and a predefined testsuite that make it easy for the user to add common testplan execution infomation - such as env var, pwd, log file, driver metadata - to the test report.

DriverLogCollector

You can specify your custom log collector using DriverLogCollector. It will attach the specified log files to the report for each driver. Example:

from testplan.common.utils import helper

log_collector = helper.DriverLogCollector(
    name="custom_log_collector",
    file_pattern=["*.log"],
    description="Driver log files",
    ignore="not_important.log",
    recursive=True,
    failure_only=False,
)
log_collector(env, result)

Required files:

• test_plan.py

test_plan.py

#!/usr/bin/env python
"""
Example demonstrating usage of testplan.common.utils.helper module.
By using the helper functions and/or the prefedined Testsuite, the user can
readily add additional information of Testplan execution to the report.
"""

import sys
from testplan import test_plan
from testplan.testing.multitest import MultiTest, testsuite, testcase
from testplan.testing.multitest.driver.app import App
from testplan.common.utils import helper


@testsuite
class MyTestsuite:
    """
    A testsuite that uses helper utilities in setup/teardown.
    """

    def setup(self, env, result):
        # Save host environment variable in report.
        helper.log_environment(result)

        # Save current path & command line arguments in report.
        helper.log_pwd(result)
        helper.log_cmd(result)

        # Save host hardware information in report.
        helper.log_hardware(result)

    @testcase
    def my_testcase(self, env, result):
        # Add your testcase here
        result.true(True)

    def teardown(self, env, result):
        """
        Attach testplan.log file in report.
        """
        helper.attach_log(result)


def before_start_fn(env, result):
    # Save host environment variable in report.
    helper.log_environment(result)

    # Save current path & command line arguments in report.
    helper.log_pwd(result)


def after_stop_fn(env, result):
    # Attach drivers' log files if the multitest failed.
    stdout_logger = helper.DriverLogCollector(
        file_pattern=["stdout*"], description="stdout"
    )
    stderr_logger = helper.DriverLogCollector(
        file_pattern=["stderr*"], description="stderr"
    )

    stdout_logger(env, result)
    stderr_logger(env, result)

    # Delete Multitest level runpath if the multitest passed.
    # This function cleans the the runpath before the exporters
    # have chance collecting the files, hence commented out.
    # helper.clean_runpath_if_passed(env, result)


@test_plan(name="Example using helper")
def main(plan):
    """
    Add a MultiTest that uses helper utilities in before_start/after_stop hooks.
    """
    plan.add(
        MultiTest(
            name="HelperTest",
            suites=[
                # This is a pre-defined testsuite that logs info to report
                helper.TestplanExecutionInfo(),
                MyTestsuite(),
            ],  # shortcut: suites=[helper.TestplanExecutionInfo()]
            environment=[App("echo", binary="/bin/echo", args=["testplan"])],
            before_start=before_start_fn,
            after_stop=after_stop_fn,
        )
    )


if __name__ == "__main__":
    sys.exit(main().exit_code)

Required files:

• test_plan.py

test_plan_metadata.py

#!/usr/bin/env python
"""
Example demonstrating the usage of on-demand Driver metadata
extraction.
"""
import sys

from testplan import test_plan
from testplan.common.utils import helper
from testplan.common.utils.context import context
from testplan.testing.multitest import MultiTest, testsuite, testcase
from testplan.testing.multitest.driver.base import (
    Direction,
    Connection,
    DriverMetadata,
)
from testplan.testing.multitest.driver.tcp import TCPServer, TCPClient


def before_start(env, result):
    """
    Extracts driver metadata before environment startup.
    """
    helper.extract_driver_metadata(env, result)


def after_start(env, result):
    """
    Accepts TCPClient connection on TCPServer side and extracts driver
    metadata after startup.
    """
    env.server.accept_connection()
    helper.extract_driver_metadata(env, result)


def metadata_extractor_server(driver: TCPServer) -> DriverMetadata:
    """
    TCPServer specific metadata extractor function.

    :param driver: TCPServer driver instance
    :return: driver name, driver class, host, and connecting port metadata
    """
    return DriverMetadata(
        name=driver.name,
        driver_metadata={
            "class": driver.__class__.__name__,
            "host": driver.host or driver.cfg.host,
        },
        conn_info=[
            Connection(
                name="data_port",
                protocol="TCP",
                identifier=driver.port or driver.cfg.port,
                direction=Direction.listening,
            )
        ],
    )


def metadata_extractor_client(driver: TCPClient) -> DriverMetadata:
    """
    TCPClient specific metadata extractor function.

    :param driver: TCPClient driver instance
    :return: driver name, driver class, host, and connecting port metadata
    """
    return DriverMetadata(
        name=driver.name,
        driver_metadata={
            "class": driver.__class__.__name__,
            "host": driver.host or driver.cfg.host,
        },
        conn_info=[
            Connection(
                name="to_server",
                protocol="TCP",
                identifier=driver.server_port or driver.cfg.port,
                direction=Direction.connecting,
            )
        ],
    )


@testsuite
class TCPSuite:
    @testcase
    def test_send_msg(self, env, result):
        """
        Simple testcase sending a message from client to server side at which it
        is received and the integrity is tested.
        """
        msg = "Hello Server!"
        msg_sent = env.client.send_text(msg)
        msg_received = env.server.receive_text(size=msg_sent)
        result.equal(msg_received, msg, "Message received on server side")


@test_plan(name="Example of Driver metadata extraction")
def main(plan):
    plan.add(
        MultiTest(
            name="Metadata extraction",
            suites=[TCPSuite()],
            environment=[
                TCPServer(
                    name="server",
                    metadata_extractor=metadata_extractor_server,
                ),
                TCPClient(
                    name="client",
                    host=context("server", "{{host}}"),
                    port=context("server", "{{port}}"),
                    metadata_extractor=metadata_extractor_client,
                ),
            ],
            before_start=before_start,
            after_start=after_start,
        )
    )


if __name__ == "__main__":
    sys.exit(not main())